package mspbots.hubspot.client.core.tickets;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.hubspot.client.core.AbstractToolsUtil;
import mspbots.hubspot.client.core.BaseApiService;
import mspbots.hubspot.common.*;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * data.connector.hubspot.core.contacts.ContactService
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/3/31
 */
@Log4j2
@Service
@RequiredArgsConstructor
public class TicketsServiceImpl extends AbstractToolsUtil implements BaseApiService {

    public Mono<JsonNode> create(Long tenantId, Map<String, Object> properties) {
        properties.put("hs_pipeline_stage", 1);
        TicketCreateBody ticketCreateBody = TicketCreateBody.builder().properties(properties).build();
        return this.hubSpotClient.post("/crm/v3/objects/tickets", ticketCreateBody)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    @Override
    public Flux<JsonNode> properties(Long tenantId) {
        return this.hubSpotClient.get("/crm/v3/properties/tickets", null)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId))
                .doOnNext(result -> {
                    Iterator<JsonNode> nodeIterator = result.withArray("results").elements();
                    Object[] columns = StreamSupport
                            .stream(Spliterators.spliteratorUnknownSize(nodeIterator, Spliterator.ORDERED), false)
                            .map(node -> node.get("name").textValue())
                            .distinct()
                            .toArray(String[]::new);
                    String properCacheKey = RedisKeys.TICKETS_ANCHOR_REDIS_KEY_PREFIX + tenantId + ":properties";
                    this.reactiveRedisTemplate.opsForSet()
                            .add(properCacheKey, columns)
                            .flatMap(res -> this.reactiveRedisTemplate.expire(properCacheKey, Duration.ofHours(8)))
                            .subscribe(saveResult -> log.debug("save cache to resids {}", saveResult));
                })
                .flatMapMany(jsonNode -> Flux.fromIterable(jsonNode.withArray("results")));
    }

    @Override
    public Flux<JsonNode> search(Long tenantId, QueryBody queryBody) {
        return this.reactiveRedisTemplate.opsForSet()
                .members(RedisKeys.TICKETS_ANCHOR_REDIS_KEY_PREFIX + tenantId + ":properties")
                .collectList()
                .flatMapMany(columns -> {
                    Set<String> stringSet = columns.parallelStream().map(Object::toString).collect(Collectors.toSet());
                    queryBody.setProperties(stringSet);
                    return this.hubSpotClient.fetch("/crm/v3/objects/tickets/search", queryBody);
                })
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    @Override
    public Mono<Void> syncProperties(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault("TICKETS-ENTITY-PROPERTIES");
        return this.properties(tenantId)
                .subscriberContext(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(result -> this.syncPropertiesResult(tenantId,
                        resultSync.getRequestId(), result))
                .then();
    }

    private void syncPropertiesResult(Long tenantId, UUID id, JsonNode resultNode) {
        this.sendEntity(RabbitKeys.PROPERTIES_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(id).action("Tickets")
                .entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    @Override
    public Mono<Void> syncData(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault("TICKETS-SYNC");
        return this.dateCacheFilterContacts(tenantId)
                .subscriberContext(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(result -> this.syncResult(tenantId, resultSync.getRequestId(), result))
                .then();
    }

    private void syncResult(Long tenantId, UUID id, JsonNode resultNode) {
        Long entityId = Long.parseLong(resultNode.get("id").textValue());
        log.debug("Tickets result id {}", resultNode.get("id").textValue());
        this.sendEntity(RabbitKeys.TICKETS_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(id).id(entityId)
                .action("none").entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    private Flux<JsonNode> dateCacheFilterContacts(Long tenantId) {
        String redisAnchorKey = RedisKeys.TICKETS_ANCHOR_REDIS_KEY_PREFIX + tenantId;
        return super.readAnchorTime(redisAnchorKey)
                .flatMapMany(syncTime -> this.search(tenantId, QueryBody.withFilter(FilterExpression.builder()
                        .operator("GT").propertyName("hs_lastmodifieddate")
                        .value(syncTime + "").build())))
                .doOnNext(user -> super.recordAnchorTime(redisAnchorKey,
                        LocalDateTime.parse(user.get("updatedAt").textValue(), DateTimeFormatter.ISO_DATE_TIME)
                                .toInstant(ZoneOffset.UTC).toEpochMilli()));
    }

}
